#include <stdlib.h>
#include <time.h>
#ifdef __linux__
#include <sys/time.h>
#endif
#include <string>
#include <iostream>
#include <memory>
#include <rise/common/exmacros.h>
#include <rise/common/Log.h>
#include <rise/common/Exception.h>
#include <rise/common/DataEncoder.h>
#include <rise/common/StreamBuffer.h>
#include <rise/common/PerformanceCounter.h>
#include <staff/common/Value.h>
#include <staff/common/QName.h>
#include <staff/common/Attribute.h>
#include <staff/common/DataObject.h>
#include <staff-mq/client/Publisher.h>

#ifdef WIN32
  LARGE_INTEGER g_llFrequency = {0};
  BOOL g_bQueryResult = QueryPerformanceFrequency(&g_llFrequency);
#endif

long long GetCurrentTime()
{
#ifdef WIN32
  LARGE_INTEGER llPerf = {0};
  QueryPerformanceCounter(&llPerf);
  return llPerf.QuadPart * 1000ll / ( g_llFrequency.QuadPart / 1000ll);
#else
  struct timeval stTimeVal;
  gettimeofday(&stTimeVal, NULL);
  return stTimeVal.tv_sec * 1000000ll + stTimeVal.tv_usec;
#endif
}


int main(int nArgs, char* szArgs[])
{
  std::string sServerHost = "127.0.0.1";
  unsigned short ushServerPort = 8768;

  for (int nArg = 1; nArg < nArgs; ++nArg)
  {
    if(szArgs[nArg])
    {
      const std::string& sArg = std::string(szArgs[nArg]);
      if (sArg.substr(0, 2) == "-s")
      {
        sServerHost = sArg.substr(2);
      }
      else
      if (sArg.substr(0, 2) == "-p")
      {
        ushServerPort = static_cast<unsigned short>(atoi(sArg.substr(2).c_str()));
      }
      else
      if (sArg.substr(0, 2) == "-h")
      {
        std::cerr << "Publisher binary data transfer test.\n\n"
            << szArgs[0] << "[ -h][ -s<Server host>][ -p<port>]\n\n";
        return 0;
      }
      else
      {
        std::cerr << "unknown arg: " << sArg << "\n";
        return 1;
      }
    }
  }


  try
  {
    const int nMinMsgSize = 10240;   // min msg size: 10kb
    const int nMaxMsgSize = 1024 * 1024 * 10; // max msg size: 10mb
    const int nMsgSizeCount = 10;    // tests count
    const std::string sQueueName = "queue-bin-bench";

    const int nMsgSizeStep = (nMaxMsgSize - nMinMsgSize) / nMsgSizeCount;

    smq::Publisher tPublisher;
    tPublisher.Init("publisher-bin", sServerHost, ushServerPort);

    tPublisher.CreateQueue(sQueueName);

    for (int nCurrMsgSize = nMinMsgSize; nCurrMsgSize <= nMaxMsgSize; nCurrMsgSize += nMsgSizeStep)
    {
      staff::CDataObject tdoMsg("msg");
      tdoMsg.SetNamespaceUri("http://tempui.org/staff-mq/tests/benchmark");

      // binary data
      staff::CDataObject tdoData = tdoMsg.CreateChild("Data");
      {
        std::string sBase64Data; // encoded data
        rise::CStreamBuffer tBuff;
        tBuff.Resize(nCurrMsgSize);
        unsigned char* pDataRaw = tBuff.GetData();

        for (long i = 0; i < nCurrMsgSize; ++i)
        {
          pDataRaw[i] = i;
        }
        // encoding data to base64 string
        rise::CBase64Encoder::Encode(tBuff, sBase64Data);

        // setting data
        tdoData.SetText(sBase64Data);
      }

      // publishing into queue
      std::cout << "Publishing message into \"" << sQueueName << "\" with size = " << nCurrMsgSize << std::endl;

      tdoMsg.CreateChild("Size").SetValue(nCurrMsgSize);
      tdoMsg.CreateChild("Time").SetValue(GetCurrentTime());
      tPublisher.Publish(sQueueName, tdoMsg);
    }

    tPublisher.RemoveQueue(sQueueName);
  }
  RISE_CATCH_ALL

  return 0;
}
